package org.example;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

//消息的发送方
public class MyProducer {
    private final static String TOPIC_NAME = "test";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        设置参数
        Properties properties=new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.0.40:9092");

//        把发送的key从字符串系列化为字节数组
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        把发送的value从字符串系列化为字节数组
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        创建生产消息的客户端，参入参数
        Producer<String,String>producer=new KafkaProducer<String, String>(properties);
//        创建消息
//        key决定往哪个分区上发，value具体发的消息
        ProducerRecord<String,String>producerRecord=new ProducerRecord<String, String>(TOPIC_NAME,"test","helloKafka");

//        发送消息 得到消息发送的源数据 输出
        RecordMetadata metadata=producer.send(producerRecord).get();
        System.out.println("同步方式发送消息结果: topic-"+metadata.topic()+" |partition-"+metadata.partition()+" |offset-"+metadata.offset());
    }
}
